package com.zzy.broker.session;

import io.netty.channel.Channel;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 会话管理器 - 单例模式
 */
public class SessionManager {
    private static final SessionManager INSTANCE = new SessionManager();
    
    // ClientID -> ClientSession
    private final ConcurrentHashMap<String, ClientSession> sessions = new ConcurrentHashMap<>();
    
    // Channel -> ClientID (用于快速查找)
    private final ConcurrentHashMap<Channel, String> channelToClientId = new ConcurrentHashMap<>();
    
    private SessionManager() {}
    
    public static SessionManager getInstance() {
        return INSTANCE;
    }
    
    /**
     * 创建或更新客户端会话
     */
    public ClientSession createOrUpdateSession(String clientId, Channel channel, boolean cleanSession, 
                                               int protocolLevel, String username) {
        ClientSession session = sessions.get(clientId);
        
        if (session != null) {
            // 如果是 cleanSession，清除旧的订阅信息
            if (cleanSession) {
                System.out.println("Clean session for client: " + clientId);
                session = new ClientSession(clientId, channel, cleanSession, protocolLevel, username);
            } else {
                // 保留旧的订阅信息，只更新 Channel
                System.out.println("Reusing session for client: " + clientId);
                // 移除旧的 channel 映射
                if (session.getChannel() != null) {
                    channelToClientId.remove(session.getChannel());
                }
                session.setChannel(channel);
            }
        } else {
            System.out.println("Creating new session for client: " + clientId);
            session = new ClientSession(clientId, channel, cleanSession, protocolLevel, username);
        }
        
        sessions.put(clientId, session);
        channelToClientId.put(channel, clientId);
        
        return session;
    }
    
    /**
     * 获取客户端会话
     */
    public ClientSession getSession(String clientId) {
        return sessions.get(clientId);
    }
    
    /**
     * 根据Channel获取会话
     */
    public ClientSession getSessionByChannel(Channel channel) {
        String clientId = channelToClientId.get(channel);
        return clientId != null ? sessions.get(clientId) : null;
    }
    
    /**
     * 移除会话
     */
    public void removeSession(String clientId) {
        ClientSession session = sessions.remove(clientId);
        if (session != null && session.getChannel() != null) {
            channelToClientId.remove(session.getChannel());
            System.out.println("Session removed for client: " + clientId);
        }
    }
    
    /**
     * 移除会话（根据Channel）
     */
    public void removeSessionByChannel(Channel channel) {
        String clientId = channelToClientId.remove(channel);
        if (clientId != null) {
            sessions.remove(clientId);
            System.out.println("Session removed for client: " + clientId);
        }
    }
    
    /**
     * 添加订阅
     */
    public void addSubscription(String clientId, String topic, int qos) {
        ClientSession session = sessions.get(clientId);
        if (session != null) {
            session.addSubscription(topic, qos);
            System.out.println("Client " + clientId + " subscribed to topic: " + topic + " (QoS " + qos + ")");
        }
    }
    
    /**
     * 移除订阅
     */
    public void removeSubscription(String clientId, String topic) {
        ClientSession session = sessions.get(clientId);
        if (session != null) {
            session.removeSubscription(topic);
            System.out.println("Client " + clientId + " unsubscribed from topic: " + topic);
        }
    }
    
    /**
     * 获取订阅了指定主题的所有客户端（支持通配符匹配）
     */
    public List<ClientSession> getSubscribedSessions(String publishTopic) {
        List<ClientSession> matchedSessions = new ArrayList<>();
        
        for (ClientSession session : sessions.values()) {
            if (!session.isActive()) {
                continue; // 跳过不活跃的会话
            }
            
            for (String subscribedTopic : session.getSubscribedTopics()) {
                if (topicMatches(subscribedTopic, publishTopic)) {
                    matchedSessions.add(session);
                    break; // 一个会话只添加一次
                }
            }
        }
        
        return matchedSessions;
    }
    
    /**
     * 主题匹配（支持 MQTT 通配符）
     * + : 单层通配符
     * # : 多层通配符
     */
    private boolean topicMatches(String filter, String topic) {
        // 如果完全相同，直接匹配
        if (filter.equals(topic)) {
            return true;
        }
        
        // 分割主题层级
        String[] filterLevels = filter.split("/");
        String[] topicLevels = topic.split("/");
        
        int filterIndex = 0;
        int topicIndex = 0;
        
        while (filterIndex < filterLevels.length && topicIndex < topicLevels.length) {
            String filterLevel = filterLevels[filterIndex];
            String topicLevel = topicLevels[topicIndex];
            
            if (filterLevel.equals("#")) {
                // # 匹配所有剩余层级
                return true;
            } else if (filterLevel.equals("+")) {
                // + 匹配单个层级
                filterIndex++;
                topicIndex++;
            } else if (filterLevel.equals(topicLevel)) {
                // 精确匹配
                filterIndex++;
                topicIndex++;
            } else {
                // 不匹配
                return false;
            }
        }
        
        // 两者都到达末尾才算完全匹配
        return filterIndex == filterLevels.length && topicIndex == topicLevels.length;
    }
    
    /**
     * 获取所有活跃会话
     */
    public List<ClientSession> getActiveSessions() {
        return sessions.values().stream()
                .filter(ClientSession::isActive)
                .collect(Collectors.toList());
    }
    
    /**
     * 获取所有会话数量
     */
    public int getSessionCount() {
        return sessions.size();
    }
    
    /**
     * 清理超时会话
     */
    public void cleanupTimeoutSessions(long timeoutMillis) {
        List<String> timeoutClients = new ArrayList<>();
        
        for (Map.Entry<String, ClientSession> entry : sessions.entrySet()) {
            ClientSession session = entry.getValue();
            if (session.isTimeout(timeoutMillis) && !session.isActive()) {
                timeoutClients.add(entry.getKey());
            }
        }
        
        for (String clientId : timeoutClients) {
            removeSession(clientId);
            System.out.println("Cleaned up timeout session: " + clientId);
        }
    }
}
